package com.beep.erp.common.rx.nettyserver;

import android.os.Handler;
import android.os.Message;
import android.util.Log;

import com.beep.erp.common.bean.ResultBean;
import com.google.gson.Gson;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.concurrent.TimeUnit;

import io.netty.channel.ChannelOption;
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.CharsetUtil;
import io.reactivex.netty.channel.Connection;
import io.reactivex.netty.protocol.tcp.client.TcpClient;
import rx.Observable;

/**
 * FileName: RxNettyManager
 * description:使用 TCP 长连接
 * Author: jolly
 * Date: 2016/9/30
 */
public class RxNettyManager {

    //多长时间为请求后，发送心跳
    private Connection<String, String> mConnection;
    private String serverIP;
    private int port;
    private static final String TAG = "NettyManager";
    private String HEAT_STRING;//心跳数据
    private String LOGIN_STRING;//登陆数据
    private boolean isOnLine = false;
    private String heartAction;
    private int spacingTime = 5;
    private NettyListener mListener;

    public interface NettyListener {
        void reciveData(String action, String message);

        void onLine();

        void offLine();
    }

//    public RxNettyManager() {
//    }


    /**
     * 初始化RxNetty异步通信库
     *
     * @param serverip    服务器IP
     * @param port        通讯端口号
     * @param login       登陆数据
     * @param heartStr    心跳数据
     * @param heartAction 心跳数据Action
     * @param spacingTime 心跳间隔
     */
    public void init(String serverip, int port, String login, String heartStr, String heartAction, int spacingTime) {
        this.serverIP = serverip;
        this.port = port;
        LOGIN_STRING = login;
        HEAT_STRING = heartStr;
        this.heartAction = heartAction;
        this.spacingTime = spacingTime;
    }


    private Handler mHandler = new Handler() {
        @Override
        public void handleMessage(Message msg) {
            super.handleMessage(msg);
            mHandler.removeMessages(0);
            send(HEAT_STRING);
        }
    };

    /**
     * 客户端连接服务器
     */
    public void startServer() {
        if (LOGIN_STRING.equals("") || LOGIN_STRING == null) {
            new RuntimeException("LOGIN_STRING isEmpty please init first");
            return;
        }
        if (HEAT_STRING.equals("") || HEAT_STRING == null) {
            new RuntimeException("HEAT_STRING isEmpty  please init first");
            return;
        }
        SocketAddress socketAddress = new InetSocketAddress(serverIP, port);
        Log.e(TAG, " rxNettyClientConnect  socketAddress : " + socketAddress.toString());
        TcpClient.newClient(socketAddress).<String, String>pipelineConfigurator(pipeline -> {
                    //这部分设置一定要和服务器上设置相同，要不然会出现无法连接的现象
                    // Decoders
                    pipeline.addLast("frameDecoder", new LineBasedFrameDecoder(1024 * 1024 *
                            1024));
                    pipeline.addLast("stringDecoder", new StringDecoder(CharsetUtil.UTF_8));
                    // Encoder
                    pipeline.addLast("stringEncoder", new StringEncoder(CharsetUtil.UTF_8));
//                        pipeline.addLast("lineEncoder", new LineEncoder(LineSeparator.UNIX,
//                                CharsetUtil.UTF_8));

                })
                .channelOption(ChannelOption.SO_KEEPALIVE, true)
                //设置读取超时时间
                .readTimeOut(spacingTime + 3, TimeUnit.SECONDS)
                .createConnectionRequest()
                .subscribe(newConnection -> {
                    mConnection = newConnection;
                    mConnection.getInput().subscribe(message -> {
                        Log.e(TAG, "receive : " + message);
                        Gson gson = new Gson();
                        String action = gson.fromJson(message, ResultBean.class).action;
                        //根据心跳进行心跳保活发送数据
                        if (heartAction.equals(action))
                            mHandler.sendEmptyMessageDelayed(0, spacingTime * 1000);
                        if (mListener != null) mListener.reciveData(action, message);
                    }, e -> reconnect(e));
                }, e -> reconnect(e), () -> {
                    if (mListener != null) mListener.onLine();
                    isOnLine = true;
                    Log.e(TAG, "connect success");
                    send(LOGIN_STRING);
                });
    }

    /**
     * 断开自动重新连接
     */
    private void reconnect(Throwable e) {
        mHandler.removeCallbacksAndMessages(null);
        //延迟spacingTime秒后进行重连
        Observable.timer(spacingTime, TimeUnit.SECONDS).subscribe(l -> {
            if (mConnection != null) mConnection.closeNow();
            Log.e(TAG, "reconnect");
            if (isOnLine) {
                if (mListener != null) mListener.offLine();
            }
            isOnLine = false;
            startServer();
        });
    }


    public void setListener(NettyListener listener) {
        mListener = listener;
    }

    /**
     * 发送数据
     *
     * @param s
     */
    public boolean send(String s) {
        if (mConnection == null) return false;
        Log.w(TAG, "send : " + s);
        //在发送的时候一定要将subscribe实现，开始在使用write函数的时候由于rxjava知识不牢，导致只是实现
        //writeString函数没有实现观察者，导致根本没有发送，所以之前一直是发送失败；
        mConnection.writeString(Observable.just(s)).subscribe(v -> {
            //暂时还不知道是干啥用的
            Log.w(TAG, "send : " + v.toString());
        }, e -> {
            //发送失败，出现异常
            Log.w(TAG, "send : " + e.getMessage());
        }, () -> {
            //发送成功
            Log.w(TAG, "send : " + "ok");
        });
        return true;
    }

    /**
     * 停止服务
     */
    public void stopServer() {
        mHandler.removeCallbacksAndMessages(null);
        if (mConnection != null) {
            mConnection.closeNow();
            mConnection = null;
        }
    }
}
